6e532337da452d88ca9d817b727ff17a687b82c0,app-fabric/src/main/java/com/continuuity/internal/app/runtime/batch/MapReduceMetricsWriter.java,MapReduceMetricsWriter,reportContinuuityStats,#,104

Before Change


    // also any other metrics (including dataset metrics)
    for (String group : counters.getGroupNames()) {
      if (group.startsWith("continuuity.")) {
        String scopePart = group.substring(group.lastIndexOf(".") + 1);
        // last one should be scope
        MetricsScope scope;
        try {
          scope = MetricsScope.valueOf(scopePart);
        } catch (IllegalArgumentException e) {
          // SHOULD NEVER happen, simply skip if happens
          continue;
        }
        reportContinuuityStats(counters.getGroup(group),
                               context.getSystemMetrics(scope), scope, previousReduceStats);
      }
    }
  }

After Change


    for (String group : counters.getGroupNames()) {
      if (group.startsWith("continuuity.")) {
        String[] parts = group.split("\\.");
        String scopePart = parts[parts.length - 1];
        // last one should be scope
        MetricsScope scope;
        try {
          scope = MetricsScope.valueOf(scopePart);
        } catch (IllegalArgumentException e) {
          // SHOULD NEVER happen, simply skip if happens
          continue;
        }

        String programPart = parts[1];
        if (programPart.equals("mapper")) {
          reportContinuuityStats(counters.getGroup(group), context.getSystemMetrics(scope), scope, previousMapStats);
        } else if (programPart.equals("reducer")) {
          reportContinuuityStats(counters.getGroup(group), context.getSystemMetrics(scope), scope, previousReduceStats);
        } else {
          reportContinuuityStats(counters.getGroup(group), context.getSystemMetrics(scope), scope, previousSystemStats);
        }